package cn.xuexiyuan.flinkstudy.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Description: 使用 DataStream 实现批处理
 * @Author 左龙龙
 * @Date 21-3-18
 * @Version 1.0
 **/
public class WordCount2 {

    public static void main(String[] args) throws Exception {
        // 0. env 创建 Flink 执行环境的上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 使用 DataStream 实现批处理
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
//        // 使用 DataStream 实现流处理
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//        // 使用 DataStream 根据数据源自动选择流还是批
//        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 1. source
        DataStream<String> lines = env.fromElements(
                "JAVA C# Python C++ Flink", "Java hello world.",
                "python hello world.", "Flink hello world."
        );

        // 2.transformation
        DataStream<Tuple2<String, Integer>> counts = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] tokens = s.toLowerCase().split("\\W+");
                for (String token : tokens) {
                    if (token.length() > 0) {
                        collector.collect(new Tuple2<>(token, 1));
                    }
                }
            }
        });
        // 分组
        KeyedStream<Tuple2<String, Integer>, String> grouped = counts.keyBy(t -> t.f0);
        // 聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = grouped.sum(1);

        // 3. sink
        result.print();

        //  4.excute 启动并等待程序结束
        env.execute("WordCount2");
    }

}
